package org.logsin37.tool.eurekamanager.infra.repository.impl;

import java.util.Collections;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.logsin37.tool.eurekamanager.domain.entity.Eureka;
import org.logsin37.tool.eurekamanager.infra.constant.ErrorMessages;
import org.logsin37.tool.eurekamanager.infra.repository.EurekaRepository;
import org.springframework.data.redis.core.ReactiveHashOperations;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
 * Eureka RepositoryI mpl
 *
 * @author logsin37 2020/01/20 16:32
 */
@Repository
@Slf4j
@AllArgsConstructor
public class EurekaRepositoryImpl implements EurekaRepository {

    private final ReactiveStringRedisTemplate redisTemplate;
    private final ObjectMapper objectMapper;

    @Override
    public Flux<Eureka> listAllEureka() {
        final ReactiveHashOperations<String, String, String> hashOperations = this.redisTemplate.opsForHash();
        return hashOperations.entries(KEY)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(entry -> {
                    try {
                        return this.objectMapper.readValue(entry.getValue(), Eureka.class);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e.getMessage(), e);
                    }
                });
    }

    /**
     * 注册Eureka
     *
     * @param eureka eureka信息
     * @return 处理结果
     */
    @Override
    public Mono<Object> saveEureka(Eureka eureka) {
        Assert.notNull(eureka, String.format(ErrorMessages.NOT_NULL, "eureka info"));
        Assert.isTrue(StringUtils.isNotBlank(eureka.getCode()), String.format(ErrorMessages.NOT_NULL, "eureka code"));
        Assert.isTrue(StringUtils.isNotBlank(eureka.getAddress()), String.format(ErrorMessages.NOT_NULL, "eureka info"));
        return Mono.just(eureka)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(eu -> {
                    try {
                        return this.objectMapper.writeValueAsString(eu);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e.getMessage(), e);
                    }
                })
                .publishOn(Schedulers.elastic())
                .flatMap(json -> this.redisTemplate.opsForHash().put(KEY, eureka.getCode(), json))
                .map(res -> Collections.emptyMap());
    }

    @Override
    public Mono<Eureka> getEurekaInfoByCode(String code) {
        Assert.isTrue(StringUtils.isNotBlank(code), String.format(ErrorMessages.NOT_NULL, "eureka code"));
        final ReactiveHashOperations<String, String, String> hashOperations = this.redisTemplate.opsForHash();
        return hashOperations.get(KEY, code)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(json -> {
                    try {
                        return this.objectMapper.readValue(json, Eureka.class);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e.getMessage(), e);
                    }
                });
    }

    @Override
    public Mono<Void> removeEurekaByCode(String code) {
        Assert.isTrue(StringUtils.isNotBlank(code), String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return this.redisTemplate.opsForHash().remove(KEY, code)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .ofType(Void.class);
    }

    @Override
    public Mono<Boolean> eurekaCodeExists(String code) {
        return Mono.just(code)
                .flatMap(c -> {
                    if(StringUtils.isBlank(c)) {
                        return Mono.just(false);
                    } else {
                        return this.redisTemplate.opsForHash().hasKey(KEY, c);
                    }
                });
    }
}
